/*
 * =====================================================================================
 *
 *       Filename:  cl_thread_pool.cxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  04/23/2011 07:27:14 AM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Chen Ken (Ken), zlchen.ken@gmail.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef __GXX_EXPERIMENTAL_CXX0X__
#define __GXX_EXPERIMENTAL_CXX0X__
#endif

#include "cl_thread_pool.hxx"

namespace neo
{
using namespace std;

////////////
CLThreadPool::CLThreadPool(int num, CLThreadPool::SchedulePolicy policy)
  : m_policy(policy),
  m_nextThr(0)
{
  // If num is equal 0, decide the read pool size according to
  // the cpu number in the system.
  if((num <= 0) && ((num = clOnlineProcessorNum()) <= 0))
  {
    num = 1;
  }

  if(policy == CLThreadPool::ROUND_ROBIN)
  {
    // each thread has its own job queue
    for(int j = 0; j < num; ++j)
    {
      m_jobQueues.push_back(move(JobQueueUPtr(new JobQueue)));
      m_thrs.push_back(move(PoolThreadUPtr(new CLPoolThread(j, this))));
    }
  }
  else
  {
    // all threads share one job queue 
    m_jobQueues.push_back(move(JobQueueUPtr(new JobQueue)));
    for(int i = 0; i < num; ++i)
    {
      m_thrs.push_back(move(PoolThreadUPtr(new CLPoolThread(0, this))));
    }
  }
}

void CLThreadPool::start(void) 
{
  size_t siz = m_thrs.size(); 
  for(unsigned i = 0; i < siz; ++i) 
  {
    m_thrs[i]->start();
  }
}

void CLThreadPool::run(const CLCallbackFunctor& job) 
{
  int nextThr = 0;
  if(m_policy == CLThreadPool::ROUND_ROBIN)
  {
    nextThr = (m_nextThr++)%(m_thrs.size());
  }
  m_jobQueues[nextThr]->push(job);
}

// not multiple threads safe.
void CLThreadPool::shutdown(void) 
{
  size_t siz = m_thrs.size(); 
  for(unsigned i = 0; i < siz; ++i)
  {
    if(m_policy == CLThreadPool::ROUND_ROBIN)
    {
      m_jobQueues[i]->push(bind(&CLPoolThread::shutdown, m_thrs[i].get()));
    }
    else
    {
      m_jobQueues[0]->push(bind(&CLPoolThread::shutdown, m_thrs[i].get()));
    }
  }

  for(unsigned j = 0; j < siz; ++j)
  {
    m_thrs[j]->join();
  }
}

} // neo

